package com.huawei.bigdate.hbaseEs.service;

import com.huawei.bigdate.hbaseEs.InitParameters;
import com.huawei.bigdate.hbaseEs.security.LoginUtil;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Properties;


@Component
public class HBaseSearch {
    private static final Log LOG = LogFactory.getLog(HBaseSearch.class.getName());
    private static final String ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL = "zookeeper/hadoop.hadoop.com";

    private static TableName tableName = null;
    private static Connection conn = null;
    private static Table table = null;
    private static Configuration conf = null;
    /*加载application.properties配置文件参数*/
    public void login() {
        //创建配置文件对象
        conf = HBaseConfiguration.create();

        //加载HDFS/HBase服务端配置，用于客户端与服务端对接
        System.out.println("----------------");
        conf.addResource(new Path(HBaseSearch.class.getClassLoader().getResource(InitParameters.hdfsSite).getPath()));
        conf.addResource(new Path(HBaseSearch.class.getClassLoader().getResource(InitParameters.hbaseSite).getPath()));
        conf.addResource(new Path(HBaseSearch.class.getClassLoader().getResource(InitParameters.coreSite).getPath()));

        try {
            //安全登录,请根据实际情况：安全模式需要安全登录；普通模式不需要
            if (User.isHBaseSecurityEnabled(conf)) {
                String krb5ConfFile = ESSearch.class.getClassLoader().getResource(InitParameters.krb5File).getPath();
                String userKeytab = ESSearch.class.getClassLoader().getResource(InitParameters.userKeytabFile).getPath();
                //配置ZooKeeper认证信息。ZooKeeper为HBase集群中各进程提供分布式协作服务
                LoginUtil.setJaasFile(InitParameters.userName,InitParameters.userKeytabFile);
                LoginUtil.setZookeeperServerPrincipal(ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL);
                LoginUtil.login(InitParameters.userName, userKeytab, krb5ConfFile, conf);
            }
            try {
                //创建Connection对象,用于连接 HBase服务器 和 ZooKeeper，Connection也提供实例化Admin和Table对象的方法。
                conn = ConnectionFactory.createConnection(conf);
            } catch (IOException e) {
                LOG.error("Failed to createConnection because ", e);
            }
        } catch (IOException e) {
            LOG.error("Failed to login because ", e);
        }
    }
    public List<String> getHbaseData(List<String> list, String testTableName) {
        //查询交易信息
        return  getDate(list, testTableName);

    }
    public static List<String> getDate(List<String> list, String testTableName) {
        List<String> imageList = new ArrayList<>();
        LOG.info("-----------Entering HBase test-------------------");
        try {
            tableName = TableName.valueOf(testTableName);//修改“testTableName”为实际表名
            table = conn.getTable(tableName);

            Get get = new Get(Bytes.toBytes(list.get(0)));
            get.addColumn(Bytes.toBytes("base"), Bytes.toBytes("cf1"));

             Result result = table.get(get);
            for (Cell cell : result.rawCells()) {
                LOG.info(Bytes.toString(CellUtil.cloneRow(cell)) + ":"
                        + Bytes.toString(CellUtil.cloneFamily(cell)) + ","
                        + Bytes.toString(CellUtil.cloneQualifier(cell)) + ","
                        + Bytes.toString(CellUtil.cloneValue(cell)));
                 imageList.add(Bytes.toString(CellUtil.cloneValue(cell)));
            }
        } catch (IOException e) {
            LOG.error("HBase test failed.", e);
        } catch (Exception e) {
            LOG.error("HBase test failed.", e);
        } finally {
            if (table != null) {
                try {
                    // 关闭table对象
                    table.close();
                } catch (IOException e) {
                    LOG.error("Failed to close table ", e);
                }
            }
        }
        LOG.info("Exiting test.");
        LOG.info("-----------finish HBase -------------------");
        return imageList;
    }
}